package com.example.latedata;

import com.example.beans.Event;
import com.example.beans.UrlViewCount;
import com.example.source.ClickSource;
import com.example.window.UrlViewCountExample_AggAndFullWindow;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;


/**
 *  
 *  * @projectName flink-1.13.2-study
 *  * @title     ProcessLateDataExample   
 *  * @package    com.example.latedata  
 *  * @description    添加侧输出流方式处理迟到数据  
 *  * @author hjx     
 *  * @date   2022-3-29 11:15  
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *  
 */
public class ProcessLateDataExample {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取数据源
        // 读取 socket 文本流
        SingleOutputStreamOperator<Event> dataStreamSource =
                env.socketTextStream("cdh1", 7777)
                        .map(new MapFunction<String, Event>() {
                            @Override
                            public Event map(String value) throws Exception {
                                String[] fields = value.split(" ");
                                return new Event(fields[0].trim(), fields[1].trim(),
                                        Long.valueOf(fields[2].trim()));
                            }
                        });

        // 打印数据生成过程
        dataStreamSource.print("dataStreamSource ");

        // 方式一：设置 watermark 延迟时间，2 秒钟
        SingleOutputStreamOperator<Event> streamWithWaterMark = dataStreamSource.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        }));

        // 定义侧输出流标签
        OutputTag<Event> lateTag = new OutputTag<Event>("late") {
        };

        // 处理数据
        /*
        用一个 AggregateFunction 来实现增量聚合，每来一个数据就计数加一；得到的结
        果交给 ProcessWindowFunction，结合窗口信息包装成我们想要的 UrlViewCount，最终输出统计结果。
         */
        SingleOutputStreamOperator<UrlViewCount> result = streamWithWaterMark.keyBy(data -> data.url)
                // 滚动事件时间窗口, 窗口大小为 10s
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 方式二：允许窗口处理迟到数据，设置 1 分钟的等待时间
                .allowedLateness(Time.minutes(1))
                // 方式三：将最后的迟到数据输出到侧输出流
                .sideOutputLateData(lateTag)
                .aggregate(new UrlViewCountAgg(), new UrlViewCountFullWindow());

        /*
        nc –lk 7777

        Alice, ./home, 1000
        Alice, ./home, 2000
        Alice, ./home, 10000
        Alice, ./home, 9000
        Alice, ./cart, 12000
        Alice, ./prod?id=100, 15000
        Alice, ./home, 9000
        Alice, ./home, 8000
        Alice, ./prod?id=200, 70000
        Alice, ./home, 8000
        Alice, ./prod?id=300, 72000
        Alice, ./home, 8000
        Alice, ./home, 80000
        Alice, ./home, 9000
         */

        result.print("result ");
        result.getSideOutput(lateTag).print("late ");

        // 执行程序
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 自定义增量聚合函数
    // 统计不同 url 的访问次数
    private static class UrlViewCountAgg implements AggregateFunction<Event, Long, Long> {

        // 初始化累加器
        @Override
        public Long createAccumulator() {
            return 0L;
        }


        // 来一条数据，增加 1
        @Override
        public Long add(Event value, Long accumulator) {
            return accumulator + 1;
        }


        // 直接返回累加的结果
        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return null;
        }
    }

    // 自定义窗口处理函数，只需要包装窗口信息
    // 输入是 不同 url 的访问次数, 输出是 UrlViewCount 封装好javabean, key 是字段 url, 窗口是 时间窗口
    private static class UrlViewCountFullWindow extends ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow> {
        /**
         * 计算窗口的值，并输出none或多个元素。
         * Evaluates the window and outputs none or several elements.
         *
         * @param s        The key for which this window is evaluated. 此窗口计算的键。
         * @param context  The context in which the window is being evaluated. 在其中计算窗口的上下文。
         * @param elements The elements in the window being evaluated. 正在求值的窗口中的元素。
         * @param out      A collector for emitting elements. 发射元素的收集器。
         * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
         */
        @Override
        public void process(String s, Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception {

            // 获取当前窗口的开始和结束时间
            long start = context.window().getStart();
            long end = context.window().getEnd();

            // 提取出当前窗口 key(url) 对应的次数 count , 迭代器就是保存上一个聚合函数中统计出的每个 url 的访问次数
            Long count = elements.iterator().next();
            // 将最终结果进行封装，输出
            out.collect(new UrlViewCount(s, count, start, end));

        }
    }

}
